package com.wanglei.project

import org.apache.spark.{SparkConf, SparkContext}

object Spark03_top10 {

  def main(args: Array[String]): Unit = {
    /**
     * top10热门品类
     * 品类id -> (点击数量，下单数量，支付数量)
     */

    val conf = new SparkConf().setMaster("local[2]").setAppName("top")
    val sc = new SparkContext(conf)

    // 
    val actionRdd = sc.textFile("datas/user_visit_action.txt")


    val flatRdd = actionRdd.flatMap(
      action => {

        val datas = action.split("_")
        if (datas(6) != "-1") {
          List((datas(6), (1, 0, 0)))
        } else if (datas(8) != "null") {
          val cids = datas(8).split(",")
          cids.map(id => (id, (0, 1, 0)))
        } else if (datas(11) != "null") {
          val cids = datas(11).split(",")
          cids.map(id => (id, (0, 0, 1)))
        } else {
          Nil
        }

      }
    )

    flatRdd.reduceByKey(
      (t1, t2) => (t1._1 + t2._1, t1._2 + t2._2, t1._3 + t2._3)
    ).sortBy(_._2, ascending = false).take(10).foreach(println(_))


    //
    sc.stop()
  }

}
